package jmind.core.redis;


import jmind.base.lang.IProperties;
import jmind.base.util.DataUtil;
import jmind.base.util.reflect.ClassUtil;
import jmind.core.serialize.Serializer;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.commands.JedisClusterCommands;

import java.io.IOException;
import java.util.*;

public class RedisCluster extends AbstractJedis {
    final JedisCluster jc;
    final Serializer serializer ;


    public RedisCluster(JedisCluster jedisCluster, Serializer serializer){
        jc=jedisCluster;
        this.serializer=serializer;
    }

    /**
     * timeout 单位秒
     * @param name
     * @param p
     */
    public RedisCluster(String name, IProperties p) {

        String[] hostAndPort = p.getProperty("redis." + name + ".host").split(",");
        Set<HostAndPort> jedisClusterNodesSet = new HashSet<HostAndPort>();
        for (String hp : hostAndPort) {
            String[] hap = hp.split(":");
            jedisClusterNodesSet.add(new HostAndPort(hap[0], Integer.parseInt(hap[1])));
        }
        GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
        genericObjectPoolConfig.setMaxIdle(DataUtil.toInt(p.getProperty("redis." + name + ".maxIdle", "10")));
        genericObjectPoolConfig.setMaxTotal(DataUtil.toInt(p.getProperty("redis." + name + ".maxTotal", "10")));
        genericObjectPoolConfig.setMinIdle(DataUtil.toInt(p.getProperty("redis." + name + ".minIdle")));
        String password=p.getProperty("redis."+name+".password");
        // 配置用秒，这里是毫秒
        int timeout=DataUtil.toInt(p.getProperty("redis." + name + ".timeout", "3")) * 1000;
        if(DataUtil.isEmpty(password)){
            jc = new JedisCluster(jedisClusterNodesSet,
                    timeout, DataUtil.toInt(p.getProperty(
                    "redis." + name + ".maxRedirections", "5")), genericObjectPoolConfig);
        }else{
            jc = new JedisCluster(jedisClusterNodesSet,timeout,
                    timeout, DataUtil.toInt(p.getProperty(
                    "redis." + name + ".maxRedirections", "5")),password, genericObjectPoolConfig);
        }

        serializer= (Serializer) ClassUtil.newInstance(p.getProperty("redis." + name + ".serializer", "jmind.core.serialize.HessianSerializer"));
    }

    public void releaseResource() {
            jc.close();


    }

    public void close(Object resource) {

    }

    public long del(String... keys) {
        return jc.del(keys);
    }

    public List<String> mget(String... key) {
        return jc.mget(key);
    }

    @Override
    public JedisClusterCommands getResource() {
        return jc;
    }


    public boolean setObject(String key, Object o)  {
        if(o==null || key==null ){
            return false;
        }
        try {
            String setObjectRet = jc.set(key.getBytes(), serializer.serialize(o));
            return "OK".equalsIgnoreCase(setObjectRet);
        } catch (IOException e) {
            logger.error("",e);
            return false;
        }



    }

    public boolean setObject(String key,int seconds, Object o) {
        if(o==null || key==null){
            return false;
        }
        try {
            String setObjectRet = jc.setex(key.getBytes(),seconds, serializer.serialize(o));
            return "OK".equalsIgnoreCase(setObjectRet);
        } catch (IOException e) {
            logger.error("",e);
            return false;
        }

    }

    public <T> T getObject(String key, Class<T> clazz) {
        byte[] bytes = jc.get(key.getBytes());
        try {
            if(bytes!=null)
           return (T) serializer.deserialize(bytes);
        } catch (Exception e) {
            logger.error("",e);
        }
        return null;
    }

    @Override
    public Map<String, Object> getObjects(Collection<String> keys, Class clazz) {
        byte[][] bytes=new byte[keys.size()][];
        int index=0;
        for(String key:keys){
             bytes[index++]=key.getBytes();
        }
        index=0;
        Map<String, Object> map= new HashMap<>();
        List<byte[]> mget = jc.mget(bytes);
        for(String key:keys){
            try {
                byte[] objectByte = mget.get(index++);
                if(objectByte!=null){
                    Object object = serializer.deserialize(objectByte);
                    map.put(key,object);
                }
            } catch (Exception e) {
                logger.error("",e);
            }
        }
        return map;
    }




}
